package shardkv

import (
	"sync"
	"time"
)

//raft节点已经同步了拉取到的分片信息，开始应用
func (kv *ShardKV) updateShard(command *ShardCommand) {
	kv.logger.Printf("updateShard: 开始处理分片更新，分片ID=%d，配置号=%d", command.Sid, command.ConfigNum)
	//配置已经过期了，说明已经应用过拉到的分片，直接忽略
	if kv.curConfig.Num != command.ConfigNum {
		kv.logger.Printf("updateShard: 配置号不匹配（当前=%d, 命令=%d），忽略更新，分片ID=%d", kv.curConfig.Num, command.ConfigNum, command.Sid)
		return
	}
	//如果不是等待拉取，说明之前也已经拉过了，退出
	if kv.Shards[command.Sid].State != NeedPull {
		kv.logger.Printf("updateShard: 分片状态不是NeedPull（当前=%v），忽略更新，分片ID=%d", kv.Shards[command.Sid].State, command.Sid)
		return
	}
	kv.logger.Printf("updateShard: 分片状态合法，允许更新本地分片")
	//直接赋值，注意此时的分片状态为NeedNotify
	kv.Shards[command.Sid] = Shard{
		DB:    make(map[string]string),
		State: command.Shard.State,
	}

	for k, v := range command.Shard.DB {
		kv.Shards[command.Sid].DB[k] = v
	}

	for clientId, seqId := range command.ClientCmdSeq {
		//只有当前节点不存在对应客户端的命令历史或者存在但是小于拉取分片中的才能更新
		if curSeq, ok := kv.ClientCmdSeq[clientId]; !ok || curSeq < seqId {
			kv.ClientCmdSeq[clientId] = seqId
		}
	}
}

//定期拉取分片
func (kv *ShardKV) pullShard() {
	for !kv.Killed() {
		if term, isLeader := kv.rf.GetState(); isLeader {
			kv.mu.RLock()
			kv.logger.Printf("pullShard: 当前为Leader（任期=%d），开始检查需要拉取的分片", term)
			startNum := kv.curConfig.Num
			prevConfig := kv.prevConfig
			kv.logger.Printf("pullShard: 当前配置号=%d，开始处理基于前序配置的分片拉取", startNum)
			var wg sync.WaitGroup
			//用旧版本是因为需要获取上个分组的地址
			for sid, gid := range prevConfig.Shards {

				needPull := kv.Shards[sid].State == NeedPull
				//状态已经变了不需要拉取
				if !needPull {
					//kv.logger.Printf("pullShard: 分片ID=%d 状态不是NeedPull（当前=%v），无需拉取", sid, kv.Shards[sid].State)
					continue
				}
				kv.logger.Printf("pullShard: 分片ID=%d 需要拉取，源GID=%d", sid, gid)
				wg.Add(1)

				go func(sid, src, startNum int, addrs []string) {
					defer wg.Done()
					index := 0
					args := ShardArgs{
						Type:      "PullShard",
						ShardId:   sid,
						ConfigNum: startNum,
					}
					kv.logger.Printf("pullShard: 启动拉取协程，分片ID=%d，源地址列表=%v", sid, addrs)
					//轮询所有节点找到leader通信
					for {
						//当前节点已经不是leader就不能再继续拉取了
						//if _, isLeader := kv.rf.GetState(); !isLeader {
						//	kv.resetNeedPull(sid, startNum)
						//	return
						//}
						//如果节点状态变化过，最好放弃本次拉取，重新拉取
						if kv.rf.StateChanged(term) {
							kv.logger.Printf("pullShard: 任期变化或不再是Leader，停止拉取分片ID=%d", sid)
							return
						}
						kv.mu.RLock()
						curNum := kv.curConfig.Num
						kv.mu.RUnlock()
						//配置已经更新了，直接退出，不应该修改其他配置版本的分片状态（理论上不会出现这样，安全检查）
						if curNum != startNum {
							kv.logger.Printf("pullShard: 配置号已更新（当前=%d, 开始=%d），停止拉取分片ID=%d", curNum, startNum, sid)
							return
						}
						kv.logger.Printf("pullShard: 尝试拉取分片ID=%d（索引=%d）", sid, index)
						reply := ShardReply{}
						end := kv.make_end(addrs[index])
						//请求拉取分片
						ok := end.Call("ShardKV.HandleShardRequests", &args, &reply)
						//拉取成功
						if ok {
							if reply.Err == OK {
								kv.logger.Printf("pullShard: 成功拉取分片ID=%d，数据条目数=%d", sid, len(reply.Shard.DB))
								command := ShardCommand{
									Type:  "updateShard",
									Shard: &reply.Shard,
									Sid:   sid,

									ConfigNum:    startNum,
									ClientCmdSeq: reply.ClientCmdSeq,
								}
								//提交到raft
								kv.rf.Start(command)
								kv.logger.Printf("pullShard: 分片ID=%d的更新命令已提交到Raft", sid)
								//需要重置，因为有可能提交的日志丢失需要重新发送
								return
								//	这一块收到重复也不能直接修改自身状态，因为自己是主动拉取方，肯定自己收到了才会通知对方删除，对方收到删除
								//并成功删除之后才会进入下一个阶段。
							} else if reply.Err == ErrDuplicate {
								kv.logger.Printf("pullShard: 分片ID=%d已处理过删除通知（ErrDuplicate），停止尝试", sid)
								//说明之前已经处理过了，理论上来说就是进入了下一个配置
								return
							}
						}
						//尝试下一个节点
						index = (index + 1) % len(addrs)
						kv.logger.Printf("pullShard: 拉取分片ID=%d失败，ok=%v，错误=%v，尝试下一个节点",
							sid, ok, reply.Err)
					}
				}(sid, gid, startNum, prevConfig.Groups[gid])
			}
			//先释放锁再等待，这样中间完成的分片可以被临时使用用于完成挑战2
			kv.mu.RUnlock()
			kv.logger.Printf("pullShard: 等待所有分片拉取协程完成，配置号=%d", startNum)
			//等待所有分片拉取结束，才可能进入下一个拉取循环，然后等待100ms，日志应当已经应用，若未应用说明可能丢失了
			wg.Wait()
			kv.logger.Printf("pullShard: 本轮分片拉取协程全部完成，配置号=%d", startNum)
		}
		time.Sleep(PullShardTimeout)
	}
}
